import asyncio
import dataclasses
import json
import time
from logging import Logger
from typing import Optional, List, Iterable, Dict

import aiohttp
import cacheout
import requests
from aiohttp import ClientTimeout
from binance.client import Client, AsyncClient, BaseClient
from binance.exceptions import BinanceAPIException

from .contants import Exchange, OrderType, ContratType, FutureType, OptionType, PositionMarginMode, DirectionType, \
    OrderState
from .base import ExchangePublicServiceBase, ExchangePriServiceBase, ExchangeQuotesCatcher, ExchangeWSRunner, \
    InstrumentQuote, ExchangeUserDataCatcher, AsyncRequest, ExchangeErrorCode, ExchangeError
from common.utils.tools import str2float
from .entities import SimpleInstrument, Instrument, OrderBook, Ticker, IndexPair, IndexPrice, Trade, Order, Position, \
    CurrencySummary, OrderBookItem, Spot, Option, Future, AuthInfo, ApiAuthInfo


@dataclasses.dataclass
class BinanceCurrencySummary(CurrencySummary):
    account_type: str = None


@dataclasses.dataclass
class BinancePosition(Position):
    account_type: str = None


@dataclasses.dataclass
class BinanceOrder(Order):
    account_type: str = None


class BinanceAccountType:
    Spot = 'spot'
    U_Future = 'usd_future'
    C_Future = 'coin_future'
    Option = 'option'

    @classmethod
    def is_asset_valid(cls, asset_type):
        return asset_type in [cls.Spot, cls.U_Future, cls.C_Future, cls.Option]

    @classmethod
    def kind2account_type(cls, inst_name: str, kind):
        if kind == ContratType.Spot:
            return cls.Spot
        if kind == ContratType.Option:
            return cls.Option
        if kind == ContratType.Future:
            if not inst_name:
                return
            inst_info = inst_name.split('_')
            if inst_info[0].endswith('USD'):
                return BinanceAccountType.C_Future
            return BinanceAccountType.U_Future

    @classmethod
    def account_type2kind(cls, account_type):
        if account_type == cls.Spot:
            return ContratType.Spot
        if account_type == cls.U_Future:
            return ContratType.Future
        if account_type == cls.C_Future:
            return ContratType.Future
        if account_type == cls.Option:
            return ContratType.Option

    @classmethod
    def all(cls):
        return cls.Spot, cls.U_Future, cls.C_Future, cls.Option

    @classmethod
    def kind2account_types(cls, kind):
        if kind is None:
            return cls.all()
        if kind == ContratType.Spot:
            return [cls.Spot]
        if kind == ContratType.Future:
            return [cls.C_Future, cls.U_Future]
        if kind == ContratType.Option:
            return [cls.Option]
        return []


class BinanceClient(Client):

    def _request_api(
        self, method, path: str, signed: bool = False, version=BaseClient.PUBLIC_API_VERSION, **kwargs
    ):
        try:
            return super()._request_api(method, path, signed=signed, version=version, **kwargs)
        except Exception as e:
            raise BinanceTools.transfer_error(e, BinanceAccountType.Spot)

    def _request_futures_api(
        self, method, path: str, signed: bool = False, version=1, **kwargs
    ):
        try:
            return super()._request_futures_api(method, path, signed=signed, version=version, **kwargs)
        except Exception as e:
            raise BinanceTools.transfer_error(e, BinanceAccountType.U_Future)

    def _request_futures_coin_api(
        self, method, path: str, signed: bool = False, version=1, **kwargs
    ):
        try:
            return super()._request_futures_coin_api(method, path, signed=signed, version=version, **kwargs)
        except Exception as e:
            raise BinanceTools.transfer_error(e, BinanceAccountType.C_Future)

    def _request_options_api(
        self, method, path: str, signed: bool = False, **kwargs
    ):
        try:
            return super()._request_options_api(method, path, signed=signed, **kwargs)
        except Exception as e:
            raise BinanceTools.transfer_error(e, BinanceAccountType.Option)

    def futures_reset_order(self, **params):
        return self._request_futures_api('put', 'order', True, data=params)

    def futures_coin_reset_order(self, **params):
        return self._request_futures_coin_api("put", "order", True, data=params)


class BinanceTools:
    EXCHANGE_COMMON_ERROR_MAP = {
        'default': ExchangeErrorCode.UnknownError,

        # 服务相关错误 (10000-10999)
        -1000: ExchangeErrorCode.ServerInternalError,  # Internal error
        -1001: ExchangeErrorCode.ServerUnavailable,  # Disconnected
        -1003: ExchangeErrorCode.ServerRateLimit,  # Too many requests
        -1021: ExchangeErrorCode.InvalidTimestamp,  # Invalid or out-of-sync timestamp
        -1022: ExchangeErrorCode.SignatureInvalid,  # Signature verification failed

        # 认证相关错误 (11000-11999)
        -2014: ExchangeErrorCode.ApiKeyInvalid,  # API key format invalid
        -2015: ExchangeErrorCode.PermissionDenied,  # Invalid API key, or incorrect permissio
    }
    ACCOUNT_TYPES_ERROR_MAP = {
        BinanceAccountType.Spot: {
            **EXCHANGE_COMMON_ERROR_MAP,
            # 请求参数相关错误 (12000-12999)
            -1100: ExchangeErrorCode.InvalidFormat,  # Illegal characters found in a parameter
            -1101: ExchangeErrorCode.MissingParameter,  # Missing mandatory parameter
            -1102: ExchangeErrorCode.InvalidParameter,  # Invalid parameter combination
            -1121: ExchangeErrorCode.InvalidSymbol,  # Invalid symbol

            # 交易相关错误 (13000-13999)
            -2010: ExchangeErrorCode.OrderRejected,  # New order rejected
            -2011: ExchangeErrorCode.OrderCanceled,  # Order does not exist
            -2013: ExchangeErrorCode.OrderNotExist,  # Order does not exist
            -2019: ExchangeErrorCode.MarginInsufficient,  # Margin insufficient
            -2021: ExchangeErrorCode.PriceProtectionTriggered,  # Order would trigger immediately
            -2022: ExchangeErrorCode.PostOnlyOrderRejected,  # Post Only order rejected due to price not satisfying conditions

            # 网络与其他错误 (15000-15999)
            -1013: ExchangeErrorCode.InvalidPrice,  # Invalid price
            -2016: ExchangeErrorCode.UnknownError,  # No matching error code found
        },
        # coin future与 usd futures错误码共用
        BinanceAccountType.U_Future: {
            # 请求参数相关错误 (12000-12999)
            -1100: ExchangeErrorCode.InvalidFormat,  # Illegal characters found in a parameter
            -1101: ExchangeErrorCode.MissingParameter,  # Missing mandatory parameter
            -1102: ExchangeErrorCode.InvalidParameter,  # Invalid parameter combination
            -1121: ExchangeErrorCode.InvalidSymbol,  # Invalid symbol

            # 交易相关错误 (13000-13999)
            -2010: ExchangeErrorCode.OrderRejected,  # New order rejected
            -2011: ExchangeErrorCode.OrderCanceled,  # Order does not exist
            -2013: ExchangeErrorCode.OrderNotExist,  # Order does not exist
            -2019: ExchangeErrorCode.MarginInsufficient,  # Margin insufficient
            -2021: ExchangeErrorCode.PriceProtectionTriggered,  # Order would trigger immediately
            -5022: ExchangeErrorCode.PostOnlyOrderRejected,
            -2022: ExchangeErrorCode.PostOnlyOrderRejected,
            # Post Only order rejected due to price not satisfying conditions
            -4061: ExchangeErrorCode.AccountInLiquidation,  # Liquidation triggered
            -4062: ExchangeErrorCode.InvalidParameter,  # Invalid or improper reduceOnly value
            -4164: ExchangeErrorCode.ExceedMaxPosition,  # Position size exceeds limit

            # 网络与其他错误 (15000-15999)
            -1013: ExchangeErrorCode.InvalidPrice,  # Invalid price
            -2016: ExchangeErrorCode.UnknownError,  # No matching error code found
        },
        BinanceAccountType.Option: {
            # 请求参数相关错误 (12000-12999)
            -1100: ExchangeErrorCode.InvalidFormat,  # Illegal characters found in a parameter
            -1101: ExchangeErrorCode.MissingParameter,  # Missing mandatory parameter
            -1102: ExchangeErrorCode.InvalidParameter,  # Invalid parameter combination
            -1121: ExchangeErrorCode.InvalidSymbol,  # Invalid symbol
            -2008: ExchangeErrorCode.UnsupportedSymbol,  # Options contract not supported

            # 交易相关错误 (13000-13999)
            -2022: ExchangeErrorCode.PostOnlyOrderRejected,  # Post Only order rejected
            -2023: ExchangeErrorCode.InvalidOrderType,  # Unsupported order type for options
            -2025: ExchangeErrorCode.MarginInsufficient,  # Margin insufficient for options trade
            -2026: ExchangeErrorCode.OrderRejected,  # Order rejected by exchange
            -2027: ExchangeErrorCode.InvalidPrice,  # Invalid price for options trade
            -2028: ExchangeErrorCode.InvalidQuantity,  # Invalid quantity for options trade
            -2031: ExchangeErrorCode.OrderNotExist,  # Order does not exist
            -2032: ExchangeErrorCode.AccountInLiquidation,  # Account in liquidation due to options trade
            -2033: ExchangeErrorCode.AccountInRisk,  # Account at risk due to options trade
            -2034: ExchangeErrorCode.PriceProtectionTriggered,  # Order rejected due to price protection mechanism

            # 网络与其他错误 (15000-15999)
            -1013: ExchangeErrorCode.InvalidPrice,  # Invalid price
            -2016: ExchangeErrorCode.UnknownError,  # No matching error code found
        }
    }

    @classmethod
    def transfer_error(cls, error: Exception, account_type=None):
        account_type = BinanceAccountType.U_Future if account_type == BinanceAccountType.C_Future else account_type
        if not isinstance(error, BinanceAPIException):
            return error
        if account_type in cls.ACCOUNT_TYPES_ERROR_MAP:
            error_code_map = cls.ACCOUNT_TYPES_ERROR_MAP[account_type]
        else:
            raise error
        error_code = error_code_map.get(error.code,) or error_code_map.get('default')
        return error_code.to_error()

    @classmethod
    def transfer_summary_data(cls, summary_data, account_type, is_cross=False):
        if account_type in [BinanceAccountType.U_Future, BinanceAccountType.C_Future]:
            margin_balance = str2float(summary_data['marginBalance'])
            return BinanceCurrencySummary(
                margin_currency=summary_data['asset'],
                margin_balance=margin_balance,
                account_type=account_type,
                equity=str2float(summary_data['marginBalance']),
                mm=str2float(summary_data['maintMargin']),
                im=str2float(summary_data['initialMargin']),
                mm_ratio=str2float(summary_data['maintMargin'], 0) / margin_balance if margin_balance else 0,
                im_ratio=str2float(summary_data['initialMargin'], 0) / margin_balance if margin_balance else 0,
                is_cross=is_cross and summary_data['marginAvailable'],
            )
        if BinanceAccountType.Option == account_type:
            greeks = summary_data.get('greeks', {})
            return BinanceCurrencySummary(
                margin_currency=summary_data['asset'],
                account_type=BinanceAccountType.U_Future,
                equity=str2float(summary_data['equity'], 0),
                margin_balance=str2float(summary_data['marginBalance'], 0),
                im=str2float(summary_data['locked'], 0), mm_ratio=0, mm=0,
                delta=str2float(greeks.get('delta'), 0),
                gamma=str2float(greeks.get('gamma'), 0),
                theta=str2float(greeks.get('theta'), 0),
                vega=str2float(greeks.get('vega'), 0),
            )
        if BinanceAccountType.Spot == account_type:
            total = str2float(summary_data['free'], 0) + str2float(summary_data['locked'], 0)
            return BinanceCurrencySummary(
                base_currency=summary_data['asset'],
                margin_currency=summary_data['asset'],
                account_type=BinanceAccountType.Spot,
                equity=total,
                margin_balance=total,
                im=str2float(summary_data['locked']),
                im_ratio=str2float(summary_data['locked']) / total if total else 0,
                mm=0, mm_ratio=0,
            )

    @classmethod
    def transfer_order_type(cls, order_type):
        if order_type == 'LIMIT':
            return OrderType.Limit
        elif order_type == 'MARKET':
            return OrderType.Market
        else:
            return OrderType.Unknown

    @classmethod
    def transfer_order_state(cls, order_state):
        if order_state in {'NEW', 'PARTIALLY_FILLED'}:
            return OrderState.Open
        if order_state == 'FILLED':
            return OrderState.Filled
        if order_state in {'CANCELED'}:
            return OrderState.Cancelled
        if order_state in {'EXPIRED'}:
            return OrderState.Closed
        return OrderState.Unknown

    @classmethod
    def transfer_order_data(cls, order_data, account_type=None) -> BinanceOrder:
        return BinanceOrder(
            order_id=order_data['orderId'], instrument=order_data['symbol'],
            amount=str2float(order_data['origQty']),
            price=str2float(order_data['price']),
            time_in_force=order_data['timeInForce'],
            filled_amount=str2float(order_data['executedQty'], 0),
            avg_price=str2float(order_data['avgPrice'], 0),
            direction=DirectionType.Buy if order_data['side'] == 'BUY' else DirectionType.Sell,
            order_type=cls.transfer_order_type(order_data['type']),
            label=order_data.get('clientOrderId'),
            state=cls.transfer_order_state(order_data['status']),
            update_timestamp=order_data['updateTime'],
            account_type=account_type,
        )

    @classmethod
    def transfer_stream_order_data(cls, stream_order_data, account_type=None, timestamp=None) -> BinanceOrder:
        return BinanceOrder(
            order_id=stream_order_data['i'], instrument=stream_order_data['s'],
            amount=str2float(stream_order_data['q']),
            price=str2float(stream_order_data['p']),
            filled_amount=str2float(stream_order_data['z']),
            avg_price=str2float(stream_order_data['ap'], 0),
            direction=DirectionType.Buy if stream_order_data['S'] == 'BUY' else DirectionType.Sell,
            order_type=cls.transfer_order_type(stream_order_data['o']),
            label=stream_order_data.get('c'),
            state=cls.transfer_order_state(stream_order_data['X']),
            update_timestamp=stream_order_data.get('T') or timestamp,
            account_type=account_type,
        )

    @classmethod
    def parse_filters_for_instrument(cls, filters, instrument):
        for _filter in filters:
            if _filter.get('filterType') == 'PRICE_FILTER':
                instrument.min_tick_size = str2float(_filter['minPrice'])
                instrument.max_tick_size = str2float(_filter['maxPrice'])
                instrument.tick_size_step = str2float(_filter['tickSize'])
            elif _filter.get('filterType') == 'LOT_SIZE':
                instrument.min_trade_size = str2float(_filter['minQty'])
                instrument.trade_size_step = str2float(_filter['stepSize'])
                instrument.max_trade_size = str2float(_filter['maxQty'])

    @classmethod
    def transfer_spot_data(cls, instrument_data, ) -> Instrument:
        filters = instrument_data['filters']

        instrument = Spot(
            instrument_name=instrument_data['symbol'],
            instrument_id=instrument_data['symbol'],
            base_currency=instrument_data['baseAsset'],
            quote_currency=instrument_data['quoteAsset'],
            counter_currency=instrument_data['baseAsset'],
            index_pair=IndexPair(base=instrument_data['baseAsset'], quote=instrument_data['quoteAsset'])
        )

        BinanceTools.parse_filters_for_instrument(filters, instrument)
        return instrument

    @classmethod
    def transfer_option_data(cls, instrument_data):
        is_reversed = instrument_data['quoteAsset'] == 'USD'
        base_currency = instrument_data['underlying'].replace(instrument_data['quoteAsset'], '')

        instrument = Option(
            instrument_name=instrument_data['symbol'],
            instrument_id=instrument_data['symbol'],
            base_currency=base_currency,
            quote_currency=instrument_data['quoteAsset'],
            margin_currency=instrument_data['quoteAsset'],
            counter_currency=base_currency,
            index_pair=IndexPair(base=base_currency, quote=instrument_data['quoteAsset']),
            reversed=is_reversed,
            expire_timestamp=instrument_data['expiryDate'],
            contract_multi=1,
            strike=str2float(instrument_data['strikePrice']),
            option_type=OptionType.Call if instrument_data['side'] == 'CALL' else OptionType.Put,
        )
        BinanceTools.parse_filters_for_instrument(instrument_data['filters'], instrument)
        return instrument

    @classmethod
    def transfer_future_data(cls, instrument_data, is_reversed=False):
        filters = instrument_data['filters']

        if instrument_data['contractType'] == 'PERPETUAL' or instrument_data['symbol'].endswith('_PERP'):
            future_type = FutureType.Perpetual
        else:
            future_type = FutureType.Delivery

        instrument = Future(
            exchange=Exchange.BINANCE,
            instrument_name=instrument_data['symbol'],
            instrument_id=instrument_data['symbol'],
            base_currency=instrument_data['baseAsset'],
            quote_currency=instrument_data['quoteAsset'],
            margin_currency=instrument_data['marginAsset'],
            counter_currency=instrument_data['baseAsset'] if not is_reversed else 'USD',
            index_pair=IndexPair(base=instrument_data['baseAsset'], quote=instrument_data['quoteAsset']),
            reversed=is_reversed, future_type=future_type,
            expire_timestamp=None if future_type == FutureType.Perpetual else instrument_data['deliveryDate'],
            contract_multi=instrument_data['contractSize'] if is_reversed else 1
        )
        BinanceTools.parse_filters_for_instrument(filters, instrument)
        return instrument

    @classmethod
    def transfer_option_position_data(cls, position_data):
        return BinancePosition(
            instrument=position_data['symbol'],
            size=abs(str2float(position_data['quantity'], 0)),
            direction='buy' if position_data['side'] == 'LONG' else 'sell',
            avg_price=str2float(position_data['entryPrice'], 0),
            margin_mode=PositionMarginMode.Cross,
            account_type=BinanceAccountType.Option,
            # TODO: binance未返回当前仓位的greeks值
        )

    @classmethod
    def transfer_future_position_data(cls, position_data, ):
        size = str2float(position_data['positionAmt'], 0)
        if not size:
            return
        position_side = position_data['positionSide']
        if position_side == 'BOTH':
            direction = 'buy' if size > 0 else 'sell'
        else:
            direction = 'buy' if position_side == 'LONG' else 'sell'

        return BinancePosition(
            instrument=position_data['symbol'],
            size=abs(str2float(position_data['positionAmt'], 0)),
            direction=direction,
            avg_price=str2float(position_data['entryPrice'], 0),
            margin_mode=PositionMarginMode.Isolated if position_data.get('isolated') else PositionMarginMode.Cross,
            im=str2float(position_data['initialMargin'], 0),
            mm=str2float(position_data['maintMargin'], 0),
            side=position_side,
            account_type=BinanceAccountType.kind2account_type(position_data['symbol'], ContratType.Future),
        )


EXCHANGE_CONFIGS = {
    'test': {
        BinanceAccountType.U_Future: {
            "ws_host": "wss://fstream.binancefuture.com",
            "api_host": 'https://testnet.binancefuture.com'
        },
        BinanceAccountType.C_Future: {
            "ws_host": "wss://dstream.binancefuture.com",
            "api_host": 'https://testnet.binancefuture.com'
        },
        BinanceAccountType.Spot: {
            "ws_host": "wss://stream.testnet.binance.vision:9443",
            "api_host": 'https://testnet.binance.vision/api'
        },
    },
    'prod': {
        BinanceAccountType.U_Future: {
            "ws_host": "wss://fstream.binance.com",
            "api_host": 'https://fapi.binance.com'
        },
        BinanceAccountType.C_Future: {
            "ws_host": "wss://dstream.binance.com",
            "api_host": 'https://dapi.binance.com'
        },
        BinanceAccountType.Spot: {
            "ws_host": "wss://stream.binance.com:9443",
            "api_host": 'https://api.binance.com'
        },
    }
}


class BinanceWSRunner(ExchangeWSRunner):
    platform = Exchange.BINANCE

    async def _request(self, method, params, callback=None, *args, **kwargs) -> AsyncRequest:
        req_id = self._request_id
        payloads = {
            "id": req_id,
            "method": method,
            "params": params
        }
        request = AsyncRequest(id=req_id, payloads=payloads, callback=callback)

        self._requests[req_id] = request
        await self._client.send(json.dumps(payloads))
        return request


class BinanceWSQuotesCatcher(ExchangeQuotesCatcher, BinanceWSRunner):

    platform = Exchange.BINANCE

    def __init__(
            self,
            env,
            account_type,
            logger=None,
            currencies=None,
            margin_currencies=None,
            connect_timeout=30,
            default_expire_ts=2000,
            *args, **kwargs
    ):
        account_config = EXCHANGE_CONFIGS.get(env, {}).get(account_type, {})
        api_host = account_config.get('api_host')
        ws_host = account_config.get('ws_host')

        if not api_host or not ws_host:
            raise ValueError(f'binance api config is invalid: {api_host} {ws_host}')

        if account_type == BinanceAccountType.U_Future:
            self._instrument_info_path = f'/fapi/v1/exchangeInfo'
        elif account_type == BinanceAccountType.C_Future:
            self._instrument_info_path = f'/dapi/v1/exchangeInfo'
        else:
            raise ValueError(f'not support account type: {account_type}')

        self.currencies = currencies or ['BTC', 'ETH']

        if margin_currencies:
            self._margin_currencies = margin_currencies
        elif account_type == BinanceAccountType.C_Future:
            self._margin_currencies = ['BTC', 'ETH']
        elif account_type == BinanceAccountType.U_Future:
            self._margin_currencies = ['USDT', 'USDC']
        else:
            self._margin_currencies = []

        self._api_host = api_host
        self._api_session = None

        self._account_type = account_type

        self._default_expire_ts = default_expire_ts

        self._upgrade_lock: Optional[asyncio.Lock] = None

        ExchangeWSRunner.__init__(
            self, ws_url=f'{ws_host}/stream', connect_timeout=connect_timeout,
            default_expire_ts=default_expire_ts,
            logger=logger,
            *args, **kwargs
        )
        ExchangeQuotesCatcher.__init__(self, *args, **kwargs)

    @property
    def name(self):
        return f'eqc:bn:{self._account_type}'

    @property
    def api_session(self) -> aiohttp.ClientSession:
        if not self._loop or not self._loop.is_running():
            raise TypeError('loop is not running')
        if not self._api_session:
            self._api_session = aiohttp.ClientSession(base_url=self._api_host, loop=self._loop)
        return self._api_session

    def _run(self):
        self._upgrade_lock = asyncio.Lock(loop=self._loop)
        return super()._run()

    def is_available(self, ):
        return self._loop.is_running() and self._client

    def get_instrument_info(self, instrument) -> Instrument:
        if instrument in self._quotes and self.is_available():
            return self._quotes[instrument].inst_info

    @property
    def _request_id(self) -> int:
        with self._request_lock:
            request_id = self._base_request_id
            self._base_request_id += 1
        return request_id

    async def _on_connected(self):
        await self._upgrade_subscribe()

    async def _on_disconnect(self):
        await super()._on_disconnect()

        for quote in self._quotes.values():
            quote.clear()

        self._index_prices.clear()
        if self._upgrade_lock.locked():
            self._upgrade_lock.release()

    async def _subscribe(self, channels):
        start, step = 0, 100
        while start * step < len(channels):
            await self._request(
                'SUBSCRIBE', channels[start*step:(start+1)*step],
                callback=self._on_subscribe
            )
            start += 1

    async def _on_subscribe(self, result, request=None):
        if not request:
            self._logger.warning(f'{self.name} handle subscribe result {result} but has not reqeust')
            return
        payloads = request.payloads or {}
        channels = payloads.get('params', [])

        if isinstance(channels, Iterable):
            self._subscribed_set.update(set(channels))
        self._logger.info(f'{self.name} current subscription cnt is {len(self._subscribed_set)}')

    async def _unsubscribe(self, channels):
        start, step = 0, 10
        while start * step < len(channels):
            await self._request(
                'UNSUBSCRIBE', {'channels': channels[start*step:(start+1)*step]},
                callback=self._on_unsubscribe
            )
            start += 1

    async def _on_unsubscribe(self, result, request=None):
        if not request:
            self._logger.warning(f'{self.name} handle unsubscribe result {result} but has not reqeust')
            return

        payloads = request.payloads or {}
        channels = payloads.get('params', [])

        if isinstance(channels, Iterable):
            self._subscribed_set -= set(channels)
        self._logger.info(f'cancel channels: {result}')

    async def _upgrade_subscribe(self, new_inst=None):
        if not self._upgrade_lock:
            self._logger.warning(f'{self.name} is not running, can not upgrade instruments info.')

        async with self._upgrade_lock:
            self._logger.info(f'{self.name} upgrade subscribe for new inst  {new_inst or ""}')

            # 防止重复执行
            if new_inst is not None and new_inst in self._quotes:
                return

            self._logger.info(f'{self.name} query symbols from {self._instrument_info_path}')
            try:
                async with self.api_session.get(self._instrument_info_path, timeout=ClientTimeout(10)) as resp:
                    data = await resp.json()
            except Exception as e:
                self._logger.exception(f'{self.name} query inst infos failed: {e}')
            symbols = data.get('symbols', [])
            if not symbols:
                self._logger.warning(f'{self.name} get instruments info got empty, please check.')
                return

            index_pairs = set()
            active_symbols = set()

            for symbol_detail in symbols:

                if symbol_detail['baseAsset'] not in self.currencies or symbol_detail['marginAsset'] not in self._margin_currencies:
                    continue
                active_symbols.add(symbol_detail['symbol'])

                if symbol_detail['symbol'] not in self._quotes:
                    inst_info = BinanceTools.transfer_future_data(
                        symbol_detail, is_reversed=self._account_type == BinanceAccountType.C_Future)
                    self._quotes[symbol_detail['symbol']] = InstrumentQuote(inst_info)

                index_pairs.add(symbol_detail['pair'])

            for symbol in list(self._quotes.keys()):
                if symbol not in active_symbols:
                    self._quotes.pop(symbol)

            wait_subscribes = ['!contractInfo']
            for index_pair in index_pairs:
                wait_subscribes.append(f'{index_pair.lower()}@indexPrice@1s')
            if self._account_type == BinanceAccountType.U_Future:
                wait_subscribes.append('!markPrice@arr@1s')
            elif self._account_type == BinanceAccountType.C_Future:
                wait_subscribes.extend([f'{index_pair.lower()}@markPrice@1s' for index_pair in index_pairs])

            wait_subscribes = [i for i in wait_subscribes if i not in self._subscribed_set]
            self._logger.info(f'{self.name} add new subscribes: {wait_subscribes}')
            await self._subscribe(wait_subscribes)

    async def _on_message(self, message):
        stream = message.get('stream', '')
        msg_id = message.get('id')

        # request response 类型消息
        if msg_id and msg_id in self._requests:
            result = message.get('result')
            _request: AsyncRequest = self._requests.pop(msg_id)
            if _request.callback and callable(_request.callback):
                try:
                    if asyncio.iscoroutinefunction(_request.callback):
                        asyncio.ensure_future(_request.callback(result, _request))
                    else:
                        _request.callback(result, _request)
                except Exception as e:
                    self._logger.exception(f'quote catcher handle callback for {message} except: {e}')

        # 订阅类消息
        elif stream:
            data = message.get('data')
            await self._handle_subscription_msg(stream, data or {})
        else:
            self._logger.warning(f'got can not handle message: {message}')

    def get_index_price(self, pair: IndexPair):
        if self.is_available():
            key = f'{pair.base}{pair.quote}'
            return self._index_prices.get(key)

    def mark_price(self, instrument, **kwargs):
        res = self._quotes.get(instrument)
        if res:
            return res.mark_price

    async def _handle_subscription_msg(self, stream: str, data):
        try:
            if stream not in self._subscribed_set:
                self._logger.warning(f'{self.name} got stream {stream} push message but not record.')
                self._subscribed_set.add(stream)
            if not stream:
                self._logger.error(f'can not handle subscription without channel info, data is {data}')
                return

            if (stream.startswith('!markPrice@arr') or '@markPrice' in stream) and isinstance(data, list):
                for item in data:
                    symbol = item.get('s')
                    quote = self._quotes.get(symbol)
                    if not quote:
                        continue
                    quote.index_price = float(item['i'])
                    quote.mark_price = float(item['p'])

                    if isinstance(quote.inst_info, Future) and quote.inst_info.future_type == FutureType.Perpetual:
                        quote.interest_rate = float(item['r'])

            elif stream == '!contractInfo':
                self._logger.info(f'got contract update message: {data}')
                symbol = data.get('s')
                state = data.get('cs')

                # 移除已交割的合约
                if state == 'DELIVERED' and symbol in self._quotes:
                    self._quotes.pop(symbol)
                if state == 'TRADING' and symbol not in self._quotes:
                    await self._upgrade_subscribe(new_inst=symbol)
            elif stream and '@indexPrice' in stream:
                self._index_prices[data['i']] = IndexPrice(
                    price=float(data['p']), timestamp=data['E']
                )

        except Exception as e:
            self._logger.exception(f'handle subscription {stream} msg {data} except: {e}')


class BinancePublicService(ExchangePublicServiceBase):
    platform = Exchange.BINANCE

    def __init__(self, env='test', logger: Logger = None, *args, **kwargs):
        super(BinancePublicService, self).__init__(env, logger, *args, **kwargs)
        self._client = Client(testnet=env in ['test', 'greeks_test'])
        self._timeout = 5
        self._account_quote_catchers: Dict[str, BinanceWSQuotesCatcher] = {}
        self._cache = cacheout.Cache()

    @property
    def name(self):
        return 'epubs:bn'

    def init_quote_catcher(self, base_currencies, margin_currencies):
        account_types = [BinanceAccountType.U_Future, BinanceAccountType.C_Future]
        for account_type in account_types:
            if account_type in self._account_quote_catchers:
                self._logger.warning(f'account type {account_type} quote catcher has inited, can not repeat do.')
                continue

            catcher = BinanceWSQuotesCatcher(
                env=self.env, account_type=account_type,
                currencies=base_currencies, margin_currencies=margin_currencies,
                logger=self._logger
            )
            catcher.run()

            self._account_quote_catchers[account_type] = catcher

    def index_price(self, pair: IndexPair) -> IndexPrice:
        if pair.quote == 'USD':
            index_price = self._account_quote_catchers[BinanceAccountType.U_Future].get_index_price(pair)
            if index_price:
                return index_price
            rs = self._client.futures_coin_mark_price(symbol=f'{pair.base}{pair.quote}_PERP', timeout=self._timeout)
        else:
            index_price = self._account_quote_catchers[BinanceAccountType.C_Future].get_index_price(pair)
            if index_price:
                return index_price
            rs = self._client.futures_mark_price(underlying=f'{pair.base}{pair.quote}', timeout=self._timeout)
        result = rs[0] if rs else None
        return IndexPrice(
            price=float(result['indexPrice']), timestamp=int(result['time'])
        ) if result else None

    def _spot_ticker(self, inst_name):
        ticker_rs = self._client.get_symbol_ticker(symbol=inst_name)
        book_rs = self._client.get_orderbook_ticker(symbol=inst_name)
        return Ticker(
            mark_price=str2float(ticker_rs['price']),
            timestamp=int(time.time() * 1000),
            best_bid_price=float(book_rs['bidPrice']),
            best_bid_size=float(book_rs['bidQty']),
            best_ask_price=float(book_rs['askPrice']),
            best_ask_size=float(book_rs['askQty']),
        )

    def _future_ticker(self, inst_name, account_type):
        if account_type == BinanceAccountType.U_Future:
            mark_price_rs = self._client.futures_mark_price(symbol=inst_name)
            # book_rs = self._client.futures_orderbook_ticker(symbol=inst_name)
        elif account_type == BinanceAccountType.C_Future:
            mark_price_rs = self._client.futures_coin_mark_price(symbol=inst_name)[0]
            # book_rs = self._client.futures_orderbook_ticker(symbol=inst_name)
        else:
            return

        return Ticker(
            mark_price=str2float(mark_price_rs['markPrice']),
            timestamp=mark_price_rs['time'],
            index_price=str2float(mark_price_rs['indexPrice']),
            # best_bid_price=float(book_rs['bidPrice']),
            # best_bid_size=float(book_rs['bidQty']),
            # best_ask_price=float(book_rs['askPrice']),
            # best_ask_size=float(book_rs['askQty']),
        )

    def _option_ticker(self, inst_name):
        mark_price_rs = self._client.options_mark_price(symbol=inst_name)
        book_rs = self._client.options_order_book(symbol=inst_name)
        return Ticker(
            mark_price=str2float(mark_price_rs['markPrice']),
            timestamp=min(book_rs['time'], mark_price_rs['time']),
            index_price=str2float(mark_price_rs['indexPrice']),
            best_bid_price=float(book_rs['bidPrice']),
            best_bid_size=float(book_rs['bidQty']),
            best_ask_price=float(book_rs['askPrice']),
            best_ask_size=float(book_rs['askQty']),

            best_ask_iv=str2float(mark_price_rs['askIV']),
            best_bid_iv=str2float(mark_price_rs['bidIV']),
            delta=str2float(mark_price_rs['delta']),
            gamma=str2float(mark_price_rs['gamma']),
            theta=str2float(mark_price_rs['theta']),
            vega=str2float(mark_price_rs['vega']),
        )

    def ticker(self, inst_name: str, kind=None, *args, **kwargs) -> Ticker:
        if not inst_name or not kind:
            raise ValueError('inst_name and kind is required when query ticker for binance')
        account_type = BinanceAccountType.kind2account_type(inst_name, kind)
        if not account_type:
            raise ValueError(f'{inst_name} and {kind} is invalid for {self.platform}')
        if account_type == BinanceAccountType.Spot:
            return self._spot_ticker(inst_name)
        elif account_type in [BinanceAccountType.U_Future, BinanceAccountType.C_Future]:
            return self._future_ticker(inst_name, account_type)
        elif account_type == BinanceAccountType.Option:
            return self._option_ticker(inst_name)

    def mark_price(self, inst_name, kind=None):
        if not inst_name or not kind:
            raise ValueError('inst name and kind is required')
        account_type = BinanceAccountType.kind2account_type(inst_name, kind)
        catcher = self._account_quote_catchers.get(account_type)
        mark_price = catcher.mark_price(inst_name) if catcher else None
        if mark_price is not None:
            return mark_price
        if account_type == BinanceAccountType.U_Future:
            return str2float(self._client.futures_mark_price(symbol=inst_name)['markPrice'])
        if account_type == BinanceAccountType.C_Future:
            return str2float(self._client.futures_coin_mark_price(symbol=inst_name)[0]['markPrice'])
        if account_type == BinanceAccountType.Spot:
            return str2float(self._client.get_symbol_ticker(symbol=inst_name)['price'])
        if account_type == BinanceAccountType.Option:
            return str2float(self._client.options_mark_price(symbol=inst_name)['markPrice'])

    def _spot_orderbook(self, inst_name) -> OrderBook:
        orderbook_data = self._client.get_order_book(symbol=inst_name)
        return OrderBook(
            asks=[OrderBookItem(item[0], item[1]) for item in orderbook_data.get('asks') or []],
            bids=[OrderBookItem(item[0], item[1]) for item in orderbook_data.get('bids') or []],
            timestamp=int(time.time() * 1000),
        )

    def _option_orderbook(self, inst_name) -> OrderBook:
        orderbook_data = self._client.options_order_book(symbol=inst_name)
        return OrderBook(
            asks=[OrderBookItem(item[0], item[1]) for item in orderbook_data.get('asks') or []],
            bids=[OrderBookItem(item[0], item[1]) for item in orderbook_data.get('bids') or []],
            timestamp=orderbook_data['T'],
        )

    def orderbook(self, inst_name: str, kind=None, **kwargs) -> OrderBook:
        if not inst_name or not kind:
            raise ValueError('inst_name and kind is required when query orderbook for binance')
        account_type = BinanceAccountType.kind2account_type(inst_name, kind)

        if account_type == BinanceAccountType.Spot:
            orderbook_data = self._client.get_order_book(symbol=inst_name)
        elif account_type == BinanceAccountType.C_Future:
            orderbook_data = self._client.futures_coin_order_book(symbol=inst_name)
        elif account_type == BinanceAccountType.U_Future:
            orderbook_data = self._client.futures_order_book(symbol=inst_name)
        elif account_type == BinanceAccountType.Option:
            orderbook_data = self._client.options_order_book(symbol=inst_name)
        else:
            raise ValueError(f'symbol {inst_name} and kind {kind} is invalid for {self.platform}')

        timestamp = int(time.time()) * 1000
        return OrderBook(
            asks=[OrderBookItem(float(item[0]), float(item[1])) for item in orderbook_data.get('asks') or []],
            bids=[OrderBookItem(float(item[0]), float(item[1])) for item in orderbook_data.get('bids') or []],
            timestamp=orderbook_data.get('T') or timestamp,
        )

    def _get_instruments_info_from_catcher(self, account_type):
        catcher = self._account_quote_catchers.get(account_type)
        instruments_info = catcher.get_instruments() if catcher else None
        return instruments_info

    def _instruments_info(self, use_cache=True) -> List[Instrument]:
        cache_key = f'{self.platform}:instrumentInfos'
        results = self._cache.get(cache_key) if use_cache else None
        if not results:
            results = []
            try:
                # Spot
                _results = []
                if use_cache:
                    _results = self._get_instruments_info_from_catcher(BinanceAccountType.Spot)
                if not _results:
                    res = self._client.get_exchange_info()
                    symbols = res['symbols']
                    _results = [BinanceTools.transfer_spot_data(symbol) for symbol in symbols]
                results.extend(_results)

                # U_Future
                _results = []
                if use_cache:
                    _results = self._get_instruments_info_from_catcher(BinanceAccountType.U_Future)
                if not _results:
                    res = self._client.futures_exchange_info()
                    symbols = res['symbols']
                    _results = [BinanceTools.transfer_future_data(symbol, is_reversed=False) for symbol in symbols]
                results.extend(_results)

                # C_Future
                _results = []
                if use_cache:
                    _results = self._get_instruments_info_from_catcher(BinanceAccountType.C_Future)
                if not _results:
                    res = self._client.futures_coin_exchange_info()
                    symbols = res['symbols']
                    _results = [BinanceTools.transfer_future_data(symbol, is_reversed=True) for symbol in symbols]
                results.extend(_results)

                # Option
                _results = []
                if use_cache:
                    _results = self._get_instruments_info_from_catcher(BinanceAccountType.Option)
                if not _results:
                    res = self._client.options_exchange_info()
                    symbols = res['optionSymbols']
                    _results = [BinanceTools.transfer_option_data(symbol) for symbol in symbols]
                results.extend(_results)
            except requests.RequestException as e:
                self._logger.error(f'{self.name} request instruments info failed: {e}')

            except Exception as e:
                self._logger.exception(f'{self.name} get instruments info except: {e}')

            if results:
                self._cache.set(cache_key, results, ttl=3600)
        return results

    def instruments_info(self, base_currency=None, margin_currency=None, kind=None, use_cache=True, **kwargs) -> List[Instrument]:
        results = self._instruments_info(use_cache=use_cache)

        _results = []
        for result in results:
            if not result:
                continue
            if kind is not None and result.kind != kind:
                continue
            if base_currency is not None and result.base_currency != base_currency:
                continue
            if margin_currency is not None and result.margin_currency != margin_currency:
                continue
            _results.append(result)

        return _results

    def get_futures_info(self):
        results = []
        account_types = [BinanceAccountType.U_Future, BinanceAccountType.C_Future]
        # ToDo 暂时只实现了通过websocket获取
        for account_type in account_types:
            catcher = self._account_quote_catchers.get(account_type)
            if catcher:
                results.extend(catcher.get_futures_info())
        return results

    def get_future_found_rate_details(self):
        results = []
        account_types = [BinanceAccountType.U_Future, BinanceAccountType.C_Future]
        # ToDo 暂时只实现了通过websocket获取
        for account_type in account_types:
            catcher = self._account_quote_catchers.get(account_type)
            if catcher:
                results.extend(catcher.get_future_found_rate_details())
        return results

    def instrument_info(self, inst, kind=None, account_type=None) -> Instrument:
        if not kind and not account_type:
            raise ValueError('get instrument info for binance kind or account type is required')

        if kind and not account_type:
            account_type = BinanceAccountType.kind2account_type(inst, kind)
        if account_type and not kind:
            account_type = account_type
            kind = BinanceAccountType.account_type2kind(account_type)

        catcher = self._account_quote_catchers.get(account_type)

        inst_info = catcher.get_instrument_info(inst) if catcher else None
        if inst_info:
            return inst_info

        instruments = self.instruments_info(kind=kind)
        results = [inst_info for inst_info in instruments if inst_info.instrument_name == inst]
        return results[0] if results else None

    def parse_inst(self, inst_name: str, kind=None) -> Optional[SimpleInstrument]:
        pass


@dataclasses.dataclass
class ListenKey:
    key: str
    timestamp: int  # 单位为秒

    @property
    def is_expired(self):
        now = time.time()
        return not self.key or (self.timestamp < now)


class BinanceUserDataCatcher(BinanceWSRunner, ExchangeUserDataCatcher):

    def __init__(self, auth_info: ApiAuthInfo, account_type, env, logger=None, *args, **kwargs):
        self._auth_info = auth_info
        if env not in EXCHANGE_CONFIGS:
            raise ValueError(f'{self.platform} got invalid args for env: {env}')
        env_configs = EXCHANGE_CONFIGS[env]
        if account_type not in env_configs:
            raise ValueError(f'{self.platform} got invalid args for env: {env}')

        self._account_type = account_type
        self._env = env
        ws_url = f'{EXCHANGE_CONFIGS[env][account_type]["ws_host"]}'

        self._account_summaries: Dict[str, CurrencySummary] = {}
        self._positions: Dict[str, Position] = {}
        self._open_orders: Optional[Dict[str, Order]] = None  # 为空时表示未载入初始化数据，无法做增量更新

        self._has_account_update = True

        self._async_client = None

        self._listen_key: Optional[ListenKey] = None

        self._scheduler: Optional[asyncio.Task] = None

        self._auth_failed_times = 0

        BinanceWSRunner.__init__(
            self, env=env, ws_url=ws_url, logger=logger,
            *args, **kwargs
        )
        ExchangeUserDataCatcher.__init__(self, *args, **kwargs)

    @property
    def async_client(self) -> AsyncClient:
        if not self._async_client:
            self._async_client = AsyncClient(
                api_key=self._auth_info.api_key,
                api_secret=self._auth_info.api_secret,
                testnet=self._env == 'test'
            )
        return self._async_client

    async def ws_url(self):
        await self._refresh_listen_key()
        if self._listen_key and not self._listen_key.is_expired:
            return self._ws_url + f'/ws/{self._listen_key.key}'
        else:
            self._logger.error(f'{self.name} can not got ws url cause can not get listen key.')

    @property
    def name(self):
        return f'eudc:bn:{self._account_type}:{self._auth_info.key()[-8:]}'

    def account_summaries(self, margin_currencies: Optional[List] = None):
        if self.is_available() and self._account_summaries is not None:
            return list(self._account_summaries.values())

    def _should_be_run(self, connected=False):
        if connected and (not self._listen_key or self._listen_key.is_expired):
            return False

        return self._auth_failed_times < 3

    def positions(self):
        if self.is_available() and self._positions is not None:
            return list(self._positions.values())

    def open_orders(self):
        if self.is_available() and self._open_orders is not None:
            return list(self._open_orders.values())

    async def _req_summaries_positions(self):
        do_force_refresh = int(time.time()) / 60 % 5 == 0   # 每5分钟强制更新一次
        if not do_force_refresh and not self._has_account_update:
            return
        summaries_data = []
        positions_data = []
        is_cross = False

        if self._account_type == BinanceAccountType.U_Future:
            account_data = await self.async_client.futures_account()
            account_mode = await self.async_client.futures_get_multi_assets_mode()
            is_cross = account_mode['multiAssetsMargin']

            summaries_data = account_data['assets']
            positions_data = account_data['positions']
        elif self._account_type == BinanceAccountType.C_Future:
            account_data = await self.async_client.futures_coin_account()
            summaries_data = account_data['assets']
            positions_data = account_data['positions']
        elif self._account_type == BinanceAccountType.Spot:
            account_data = await self.async_client.get_account()
            summaries_data = account_data['balances']
            self._logger.info(f'binance spot summary data is {summaries_data}')

        for summary_data in summaries_data:
            summary = BinanceTools.transfer_summary_data(summary_data, self._account_type, is_cross)
            if summary:
                self._account_summaries[summary.margin_currency or summary.base_currency] = summary

        _account_summaries = {}
        for summary_data in summaries_data:
            summary = BinanceTools.transfer_summary_data(summary_data, self._account_type, is_cross)
            if summary:
                _account_summaries[summary.margin_currency or summary.base_currency] = summary
        self._account_summaries = _account_summaries

        _positions = {}
        for position_data in positions_data:
            position = None
            if self._account_type in [BinanceAccountType.U_Future, BinanceAccountType.C_Future]:
                position = BinanceTools.transfer_future_position_data(position_data,)
            elif self._account_type in [BinanceAccountType.Option]:
                position = BinanceTools.transfer_option_position_data(position_data, )
            if position:
                _positions[f'{position.instrument}:{position.side}'] = position
        self._positions = _positions
        self._has_account_update = False

    async def _req_open_orders(self):
        open_orders = {}

        if self._account_type == BinanceAccountType.U_Future:
            open_orders_data = await self.async_client.futures_get_open_orders()
        elif self._account_type == BinanceAccountType.C_Future:
            open_orders_data = await self.async_client.futures_coin_get_open_orders()
        else:
            return

        for open_order in open_orders_data:
            order = BinanceTools.transfer_order_data(open_order, account_type=self._account_type)
            open_orders[order.order_id] = order

        self._open_orders = open_orders

    async def _on_connected(self):
        # 获取listenKey
        try:
            await self._req_open_orders()
            await self._req_summaries_positions()

            # 启动定时程序定时启动任务
            if self._scheduler and not self._scheduler.done():
                self._scheduler.cancel()

            self._loop.call_later(600, self._loop.create_task, self._refresh_listen_key())

            # 启动调度器
            self._scheduler = asyncio.create_task(self._schedule(5))
            self._logger.info(f'{self.name} start timer to refresh summaries interval.')
        except Exception as e:
            self._logger.exception(f'{self.name} do connected event except: {e}')

    async def _on_disconnect(self):
        await super(BinanceUserDataCatcher, self)._on_disconnect()
        if self._scheduler and not self._scheduler.done():
            self._scheduler.cancel()

        self._scheduler = None
        self._open_orders = None
        self._positions = None
        self._account_summaries = None

    async def _schedule(self, interval=5):
        refresh_key_ts = int(time.time())

        while self.is_available():
            now = int(time.time())
            await asyncio.sleep(interval)

            try:
                await self._req_summaries_positions()
            except aiohttp.ClientError as e:
                self._logger.error(f'schedule req summaries positions data failed: {e}')

            # 每十分种刷新一次ListenKey
            if refresh_key_ts - now > 600:
                rs = await self._refresh_listen_key()
                if rs:
                    refresh_key_ts = now

    async def _refresh_listen_key(self):
        if not self._listen_key or self._listen_key.is_expired:
            try:
                key = None
                if self._account_type == BinanceAccountType.Spot:
                    key = await self.async_client.stream_get_listen_key()
                elif self._account_type == BinanceAccountType.U_Future:
                    key = await self.async_client.futures_stream_get_listen_key()
                elif self._account_type == BinanceAccountType.C_Future:
                    key = await self.async_client.futures_coin_stream_get_listen_key()
                if key:
                    self._listen_key = ListenKey(
                        key=key, timestamp=int(time.time() + 3600)  # 默认key有效期为3600秒
                    )
                return True
            except aiohttp.ClientError as e:
                self._logger.error(f'{self.name} get listen key failed: {e}')
            except Exception as e:
                self._logger.exception(f'{self.name} get steam listen key failed {e}')
                self._auth_failed_times += 1
        else:
            try:

                if self._account_type == BinanceAccountType.Spot:
                    await self.async_client.stream_keepalive(self._listen_key.key)
                elif self._account_type == BinanceAccountType.U_Future:
                    await self.async_client.futures_stream_keepalive(self._listen_key.key)
                elif self._account_type == BinanceAccountType.C_Future:
                    await self.async_client.futures_stream_keepalive(self._listen_key.key)

                self._listen_key.timestamp = int(time.time() + 3600)
                return True
            except aiohttp.ClientError as e:
                self._logger.error(f'{self.name} refresh listen key failed: {e}')
            except Exception as e:
                self._logger.exception(f'{self.name} get stream listen key failed {e}')
                self._auth_failed_times += 1
        return False

    async def _on_message(self, message):
        self._logger.info(f'{self.name} recv message {message}')
        event = message.get('e')
        if event == 'ACCOUNT_UPDATE':
            self._has_account_update = True

        elif event == 'ORDER_TRADE_UPDATE':
            steam_order_data = message.get('o')
            if self._open_orders is None:
                return
            order = BinanceTools.transfer_stream_order_data(
                steam_order_data, account_type=self._account_type, timestamp=message.get('T')
            )
            if order.state in {OrderState.Filled, OrderState.Cancelled, OrderState.Closed}:
                self._open_orders.pop(order.order_id)
            elif order.state == OrderState.Open:
                self._open_orders[order.order_id] = order
            else:
                self._logger.warning(f'{self.name} get unknown order state {steam_order_data}')
        elif event == 'listenKeyExpired':
            # 刷新key
            self._listen_key = None
            await self._refresh_listen_key()


class BinancePriService(ExchangePriServiceBase):
    platform = Exchange.BINANCE

    def __init__(self, auth_info: AuthInfo, env='test', *args, **kwargs):
        if not isinstance(auth_info, ApiAuthInfo):
            raise ValueError(f'{self.platform} private service only support `api` and `oauth` type auth info.')
        super(BinancePriService, self).__init__(auth_info, env, *args, **kwargs)
        self._auth_info: ApiAuthInfo = auth_info
        self._client = BinanceClient(api_key=auth_info.api_key, api_secret=auth_info.api_secret, testnet=env == 'test')
        self._user_data_catchers: Dict[str, BinanceUserDataCatcher] = {}
        self._cache = cacheout.Cache()

    @property
    def name(self):
        return f'epris:bn:{self._auth_info.key()[-8:]}'

    def check_auth(self, account_types=None):
        try:
            if self.env == 'prod' and (account_types is None or BinanceAccountType.Spot in account_types):
                self._client.get_account()
            if account_types is None or BinanceAccountType.C_Future in account_types:
                self._client.futures_account()
            # if account_types is None or BinanceAccountType.Option in account_types:
            #     self._client.options_account_info()
            if account_types is None or BinanceAccountType.U_Future in account_types:
                self._client.futures_coin_account()
        except Exception as e:
            raise BinanceTools.transfer_error(e)

    def ensure_catcher_run(self):
        account_types = [
            BinanceAccountType.U_Future, BinanceAccountType.C_Future, BinanceAccountType.Spot
        ]
        for account_type in account_types:
            if account_type == BinanceAccountType.Spot and self.env != 'prod':
                continue

            catcher = self._user_data_catchers.get(account_type)
            if not catcher:
                catcher = BinanceUserDataCatcher(
                    self._auth_info, account_type, self.env, logger=self._logger)
                self._user_data_catchers[account_type] = catcher
            if not catcher.is_run():
                catcher.run()

    def account_summaries_from_catcher(self, account_type):
        catcher = self._user_data_catchers.get(account_type)
        if catcher and catcher.is_available():
            return catcher.account_summaries()

    def account_summaries(self, account_types=None, **kwargs) -> List[BinanceCurrencySummary]:
        results = []

        account_types = BinanceAccountType.all() if not account_types else account_types
        for account_type in account_types:
            # 优先从websocket中获取
            account_summaries = self.account_summaries_from_catcher(account_type)
            if account_summaries is not None:
                results.extend(account_summaries)
                self._logger.info(f'{self.name} get {account_type} summary from ws: {account_summaries}')
                continue

            try:
                if account_type in [BinanceAccountType.U_Future, BinanceAccountType.C_Future]:
                    if account_type == BinanceAccountType.U_Future:
                        data = self._client.futures_account()
                        # 判断是否联合保证金模式
                        account_mode = self._client.futures_get_multi_assets_mode()
                        is_cross = account_mode['multiAssetsMargin']
                    else:
                        data = self._client.futures_coin_account()
                        is_cross = False

                    assets = data['assets']
                    for asset in assets:
                        summary = BinanceTools.transfer_summary_data(asset, account_type, is_cross=is_cross)
                        summary and results.append(summary)
                elif account_type == BinanceAccountType.Option and self.env == 'prod':
                    data = self._client.futures_coin_account()
                    assets = data['assets']
                    for asset in assets:
                        summary = BinanceTools.transfer_summary_data(asset, account_type)
                        summary and results.append(summary)
                elif account_type == BinanceAccountType.Spot and self.env == 'prod':
                    data = self._client.get_account()
                    assets = data['balances']
                    for asset in assets:
                        summary = BinanceTools.transfer_summary_data(asset, account_type)
                        summary and results.append(summary)
            except BinanceAPIException as e:
                raise BinanceTools.transfer_error(e)
            except Exception as e:
                self._logger.exception(f'{self.name} get {account_type} summary failed: {e}')

        return results

    def account_summary(self, base_currency=None, margin_currency=None, **kwargs) -> CurrencySummary:
        pass

    def positions_from_catcher(self, kind):
        if kind == ContratType.Future:
            account_types = [BinanceAccountType.U_Future, BinanceAccountType.C_Future]
        elif kind == ContratType.Option:
            account_types = [BinanceAccountType.Option]
        else:
            return

        results = []
        for account_type in account_types:
            catcher = self._user_data_catchers.get(account_type)
            data = catcher.positions() if catcher else None
            if data is None:
                return
            results.extend(data)
        return results

    def positions(self, base_currency=None, kinds=None, **kwargs) -> List[BinancePosition]:
        results = []
        if kinds is None or ContratType.Option in kinds:
            # 暂不开放option持仓查询
            # option_positions = self.positions_from_catcher(ContratType.Option)
            option_positions = []
            if option_positions is None:
                positions_data = self._client.options_positions()
                option_positions = [BinanceTools.transfer_option_position_data(data) for data in positions_data]
            results.extend(option_positions)
        if kinds is None or ContratType.Future in kinds:
            future_positions = self.positions_from_catcher(ContratType.Future)
            if future_positions is None:
                positions_data = self._client.futures_coin_account()['positions']
                results.extend([BinanceTools.transfer_future_position_data(data) for data in positions_data])
                positions_data = self._client.futures_account()['positions']
                results.extend([BinanceTools.transfer_future_position_data(data) for data in positions_data])
            else:
                results.extend(future_positions)

        return [result for result in results if result]

    def open_orders_from_catcher(self, account_type):
        catcher = self._user_data_catchers.get(account_type)
        return catcher.open_orders() if catcher else None

    def open_orders(self, kind=None, instrument=None, **kwargs) -> List[BinanceOrder]:
        results = []

        account_types = BinanceAccountType.kind2account_types(kind)
        for account_type in account_types:
            _results = self.open_orders_from_catcher(account_type)
            if _results is not None:
                results.extend(_results)
                continue

            if account_type == BinanceAccountType.U_Future:
                data = self._client.futures_get_open_orders(symbol=instrument)
            elif account_type == BinanceAccountType.C_Future:
                data = self._client.futures_coin_get_open_orders(symbol=instrument)
            elif account_type == BinanceAccountType.Spot and self.env == 'prod':
                data = self._client.get_open_orders(symbol=instrument)
            elif account_type == BinanceAccountType.Option and self.env == 'prod':
                # TODO: option暂时不请求
                continue
            else:
                continue
            _results = [BinanceTools.transfer_order_data(i, account_type=account_type) for i in data]
            results.extend(_results)
        return results

    def create_order(self, instrument, size, price, order_type=OrderType.Limit, price_type=None, reduce_only=False,
                     post_only=False, label=None, kind=None, position_side='BOTH', **kwargs):
        # TODO: 暂不支持timeInForce作为入参
        if not kind:
            raise ValueError('kind is required when create order for binance.')
        account_type = BinanceAccountType.kind2account_type(instrument, kind)
        if account_type == BinanceAccountType.Spot:
            if order_type == OrderType.Limit and post_only:
                order_type = 'LIMIT_MAKER'
            else:
                order_type = 'LIMIT' if order_type == OrderType.Limit else 'MARKET'

            res = self._client.create_order(
                symbol=instrument, side='BUY' if size > 0 else 'SELL',
                type=order_type, newClientOrderId=label,
                timeInForce='GTC', quantity=abs(size), price=price,
                reduceOnly='true' if reduce_only else 'false'
            )
            return BinanceTools.transfer_order_data(res, )
        elif account_type in [BinanceAccountType.U_Future, BinanceAccountType.C_Future]:
            side = 'BUY' if size > 0 else 'SELL'

            time_in_force = 'GTC'
            if order_type == OrderType.Limit and post_only:
                time_in_force = 'GTX'
            order_type = 'LIMIT' if order_type == OrderType.Limit else 'MARKET'

            if account_type == BinanceAccountType.U_Future:
                # 设置默认下单倍率
                self._client.futures_change_leverage(symbol=instrument, leverage=5)

                res = self._client.futures_create_order(
                    symbol=instrument, side=side, type=order_type, timeInForce=time_in_force,
                    quantity=abs(size), price=price, positionSide=position_side,
                    newClientOrderId=label,
                    reduceOnly='true' if reduce_only else 'false'
                )
            else:
                # 设置默认下单倍率
                self._client.futures_coin_change_leverage(symbol=instrument, leverage=5)
                res = self._client.futures_coin_create_order(
                    symbol=instrument, side=side, type=order_type, timeInForce=time_in_force,
                    quantity=abs(size), price=price, positionSide=position_side,
                    newClientOrderId=label,
                    reduceOnly='true' if reduce_only else 'false'
                )
            return BinanceTools.transfer_order_data(res, )
        elif account_type == BinanceAccountType.Option:
            raise ValueError('account type of option dos not support.')

    def cancel_order(self, order_id=None, kind=None, instrument=None, label=None, account_type=None, **kwargs):
        if (not instrument or not kind) and not account_type:
            raise ValueError('instrument or kind is required.')
        account_type = BinanceAccountType.kind2account_type(instrument, kind) if not account_type else account_type

        if account_type == BinanceAccountType.Spot:
            return BinanceTools.transfer_order_data(self._client.cancel_order(symbol=instrument, orderId=order_id, ))
        if account_type == BinanceAccountType.U_Future:
            return BinanceTools.transfer_order_data(self._client.futures_cancel_order(symbol=instrument, orderId=order_id, origClientOrderId=label))
        if account_type == BinanceAccountType.C_Future:
            return BinanceTools.transfer_order_data(self._client.futures_coin_cancel_order(symbol=instrument, orderId=order_id, origClientOrderId=label))
        if account_type == BinanceAccountType.Option:
            return BinanceTools.transfer_order_data(self._client.options_cancel_order(symbol=instrument, orderId=order_id, origClientOrderId=label))
        return

    def reset_order(self, order_id, price=None, size=None, instrument=None, kind=None, label=None, **kwargs):
        if not instrument or not kind:
            raise ValueError('instrument and kind is required.')
        if not price or not size:
            raise ValueError('price and size is required.')
        if not order_id and not label:
            raise ValueError('order_id or label is required.')
        try:
            side = 'BUY' if size > 0 else 'SELL'
            params = {
                'orderId': order_id, 'price': price, 'quantity': size,
                'origClientOrderId': label, 'symbol': instrument, 'side': side,
            }
            params = {key: value for key, value in params.items() if value is not None}

            account_type = BinanceAccountType.kind2account_type(instrument, kind)

            if account_type == BinanceAccountType.U_Future:
                return BinanceTools.transfer_order_data(self._client.futures_reset_order(**params))
            if account_type == BinanceAccountType.C_Future:
                return BinanceTools.transfer_order_data(self._client.futures_coin_reset_order(**params))
            if account_type in [BinanceAccountType.Spot]:
                raise ValueError('binance spot does not support reset order')
        except ExchangeError as e:
            if e.code == ExchangeErrorCode.ResetOrderBySameParams.code:
                return self.order_state(order_id=order_id, kind=kind, instrument=instrument)
            raise e

    def order_trades(self, order_id, **kwargs) -> List[Trade]:
        pass

    def get_order_from_catcher(self, kind, instrument, order_id):
        account_type = BinanceAccountType.kind2account_type(instrument, kind)
        catcher = self._user_data_catchers.get(account_type)
        if not catcher:
            return
        open_orders = catcher.open_orders()
        if open_orders:
            for open_order in open_orders:
                if open_order.instrument == instrument and order_id == open_order.order_id:
                    return open_order

    def order_state(self, order_id, kind=None, instrument=None, **kwargs):
        if not kind or not instrument:
            raise ValueError(f'kind and instrument is required when get {self.platform} order state')
        order = self.get_order_from_catcher(kind, instrument, order_id=order_id)
        account_type = BinanceAccountType.kind2account_type(instrument, kind)
        if not order:
            order_data = None
            if account_type == BinanceAccountType.Spot:
                order_data = self._client.get_order(orderId=order_id, symbol=instrument)
            elif account_type == BinanceAccountType.U_Future:
                order_data = self._client.futures_get_order(orderId=order_id, symbol=instrument)
            elif account_type == BinanceAccountType.C_Future:
                order_data = self._client.futures_coin_get_order(orderId=order_id, symbol=instrument)
            elif account_type == BinanceAccountType.Option:
                order_data = self._client.options_query_order(orderId=order_id, symbol=instrument)
            order = BinanceTools.transfer_order_data(order_data) if order_data else None
        return order

